package cn.demoncat.util.hadoop.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;

/**
 * HDFS客户端工具
 * 
 * @author 延晓磊
 *
 * @since 2019年11月13日
 */
public class HdfsUtil {

	/**
	 * 获取HDFS文件系统连接
	 * <br/>
	 * 前提：配置环境变量HADOOP_USER_NAME，来指定Hadoop用户
	 * 
	 * @param url	HFDS的NameNode地址
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2019年11月20日
	 */
	public static FileSystem getFs(String url) throws IOException {
		// 连接配置 
		Configuration conf = new Configuration();
		// NameNode连接地址，连接用户通过环境变量HADOOP_USER_NAME获取
		conf.set("fs.defaultFS", url);
		// 文件系统
		return FileSystem.get(conf);
	}
	
	/**
	 * 获取HDFS文件系统连接
	 * 
	 * @param url	HFDS的NameNode地址
	 * @param user	Hadoop用户
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2019年11月20日
	 */
	public static FileSystem getFs(String url, String user) throws IOException, InterruptedException, URISyntaxException {
		// 连接配置 
		Configuration conf = new Configuration();
		// 文件系统
		return FileSystem.get(new URI(url), conf, user);
	}
	
	/**
	 * 关闭连接
	 * 
	 * @param fs	文件系统连接
	 * 
	 * @author 延晓磊
	 *
	 * @since 2019年11月20日
	 */
	public static void close(FileSystem fs) {
		if (fs != null) {
			try {
				fs.close();
			} catch (IOException e) {
				// 关闭失败
			}
		}
	}
	
	/**
	 * 创建目录
	 * 
	 * @param fs	文件系统连接
	 * @param dir	目录（自动创建必要的父目录，已存在则忽略）
	 * 
	 * @author 延晓磊
	 *
	 * @since 2019年11月20日
	 */
	public static void mkdir(FileSystem fs, String dir) throws IllegalArgumentException, IOException {
		// 创建目录：自动创建必要的父目录，已存在则忽略
		fs.mkdirs(new Path(dir));
	}
	
	/**
	 * 列表文件
	 * 
	 * @param fs	文件系统连接
	 * @param dir	目录
	 * @param recursive	是否递归
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2019年11月20日
	 */
	public static ArrayList<LocatedFileStatus> listFile(FileSystem fs, String dir, boolean recursive) throws IllegalArgumentException, IOException {
		// 列表文件：目录，递归子目录
		RemoteIterator<LocatedFileStatus> files = fs.listFiles(new Path("/"), recursive);
		// 转换为List
		ArrayList<LocatedFileStatus> list = new ArrayList<>();
		while (files.hasNext()) {
			list.add(files.next());
		}
		return list;
	}
	
	/**
	 * 删除文件
	 * 
	 * @param fs	文件系统连接
	 * @param path	文件路径（递归删除）
	 * 
	 * @author 延晓磊
	 * @throws IOException 
	 * @throws IllegalArgumentException 
	 *
	 * @since 2019年11月20日
	 */
	public static void del(FileSystem fs, String path) throws IllegalArgumentException, IOException {
		// 删除：递归目录
		fs.delete(new Path(path), true);
	}
	
	/**
	 * 上传文件
	 * 
	 * @param fs	文件系统连接
	 * @param path	上传路径
	 * @param file	文件
	 * 
	 * @author 延晓磊
	 *
	 * @since 2019年11月20日
	 */
	public static void put(FileSystem fs, String path, byte[] file) throws IllegalArgumentException, IOException {
		// 获取输出流：路径，是否覆盖
		FSDataOutputStream out = fs.create(new Path(path), true);
		try {
			// 上传
			out.write(file);
			out.flush();
		} finally {
			IOUtils.closeStream(out);
		}
	}
	
	/**
	 * 下载
	 * 
	 * @param fs	文件系统连接
	 * @param path	文件路径
	 * @return 文件
	 * 
	 * @author 延晓磊
	 *
	 * @since 2019年11月20日
	 */
	public static byte[] get(FileSystem fs, String path) throws IOException {
		// 获取输入流：路径
		FSDataInputStream in = fs.open(new Path(path));
		try {
			// 下载
			return IOUtils.readFullyToByteArray(in);
		} finally {
			IOUtils.closeStream(in);
		}
	}

}
